package org.zjt.demo.service.imp;

import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;
import org.zjt.demo.disruptor.Handler1;
import org.zjt.demo.disruptor.Handler2;
import org.zjt.demo.disruptor.Handler3;
import org.zjt.demo.model.Trade;
import org.zjt.demo.service.IDisruptorService;
import org.zjt.demo.disruptor.TradePublisher;
import java.util.concurrent.*;

/**
 * Created by zhang on 2017/3/2.
 */
@Service
public class DisruptorServiceImp implements IDisruptorService{
    private static final Logger LOGGER =Logger.getLogger(DisruptorServiceImp.class);
    private static final int BUFFER_SIZE= 1 << 10;

    @Autowired
    private Executor myExecutor;

    //多用于并行计算和统计
    @Async
    @Override
    public void sendMessage(Integer threadCnt) throws Exception{

        ThreadFactory threadFactory=(r)-> new Thread(r);
        long beginTime=System.currentTimeMillis();
        Disruptor<Trade> disruptor2 = new Disruptor<Trade>(()-> new Trade(), BUFFER_SIZE, threadFactory, ProducerType.MULTI, new BusySpinWaitStrategy());

        //菱形操作
        //使用disruptor创建消费者组C1,C2
        //声明在C1,C2并行完事之后执行JMS消息发送操作，也就是流程走到C3
        EventHandlerGroup<Trade> handlerGroup = disruptor2.handleEventsWith(new Handler1(), new Handler2());
        handlerGroup.then(new Handler3());
        disruptor2.start();

        //生产者准备,生产是异步的。
        CountDownLatch latch=new CountDownLatch(threadCnt);
        for (int i = 0 ; i < threadCnt ; i++)
            myExecutor.execute(threadFactory.newThread(new TradePublisher(latch, disruptor2,"thread_"+(i+1))));
        latch.await();//等待生产者完事.
        disruptor2.shutdown();
        LOGGER.info("总耗时:"+((System.currentTimeMillis()-beginTime)/1000)+"秒");
    }
}
